package com.baike.controller;

import com.alibaba.fastjson.JSONObject;
import com.baike.service.FreemarkerService;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQPushConsumerLifecycleListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.context.config.annotation.RefreshScope;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import java.util.List;

import static org.apache.rocketmq.spring.support.RocketMQHeaders.TOPIC;

/**
 * @author:pjm
 */
@Component
@RocketMQMessageListener(topic = "lessons_Topic",consumerGroup = "bbbb")
//@RefreshScope //开启自动刷新配置
public class FreemarkerController implements RocketMQListener<Object>, RocketMQPushConsumerLifecycleListener {
    @Autowired
    private RedisTemplate redisTemplate;

    @Autowired
    private FreemarkerService freemarkerService;

    @Override
    public void onMessage(Object o) {

    }

    @Override
    public void prepareStart(DefaultMQPushConsumer defaultMQPushConsumer) {
        defaultMQPushConsumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                for (MessageExt msg:list) {
                    //解决幂等性问题,先判断是否有这个key
                    Boolean goodsMessage =   redisTemplate.boundHashOps("lessonsMessage").hasKey(msg.getMsgId());
                    try {
                        if (!goodsMessage){
                            System.out.println("消息重试次数" + msg.getReconsumeTimes());
                            System.out.println("消息重试主题" + msg.getTopic());
                            String s = new String(msg.getBody(),"utf-8");
                            Long lessonId = JSONObject.parseObject(s, Long.class);

                            System.out.println("消息" +lessonId);
                            //将消息id存到redis中
                            redisTemplate.boundHashOps("lessonsMessage").put(msg.getMsgId(),lessonId);
                            //调用方法
                            freemarkerService.creatPage(lessonId);
                        }else {
                            System.out.println("没进入判断");
                        }
                    } catch (Exception e) {
                        e.printStackTrace();
                        System.out.println("出现异常"+e.getMessage());
                        //重试次数
                        if(msg.getReconsumeTimes()>2){
                            consumeConcurrentlyContext.setDelayLevelWhenNextConsume(-1);
                            //手动返回消费成功
                            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                        }
                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                    }
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

    }
}
